package com.mycompany.bigdata;

import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Properties;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.goldengate.atg.datasource.AbstractHandler;
import com.goldengate.atg.datasource.DsConfiguration;
import com.goldengate.atg.datasource.DsEvent;
import com.goldengate.atg.datasource.DsOperation;
import com.goldengate.atg.datasource.DsTransaction;
import com.goldengate.atg.datasource.GGDataSource.Status;
import com.goldengate.atg.datasource.adapt.Col;
import com.goldengate.atg.datasource.adapt.Op;
import com.goldengate.atg.datasource.adapt.Tx;
import com.goldengate.atg.datasource.meta.ColumnMetaData;
import com.goldengate.atg.datasource.meta.DsMetaData;
import com.goldengate.atg.datasource.meta.TableMetaData;
import com.goldengate.atg.datasource.meta.TableName;
import com.goldengate.atg.util.ConfigException;

public class SampleHandlerFlume extends AbstractHandler {
  final private static Logger logger = LoggerFactory
      .getLogger(SampleHandlerFlume.class);

  /** count total number of operations */
  private long numOps = 0;

  /** count total number of transactions */
  private long numTxs = 0;

  private String regularFileName = "default_flume_debug.txt";

  /** At least 2 hosts are required */
  private String flumeHosts = "";// 10.20.13.103:50001,10.20.13.103:50002,10.20.1h3.103:50003
  private String clientType = "default_loadbalance";
  private String hostSelector = "round_robin";
  private boolean backOff = false;// Specifies whether the client should
  private int maxBackOff = 0;
  private int batchSize = 100;
  private int connectTimeout = 20000;
  private int requestTimeout = 20000;
  private int sendThread = 1;
  private int sendThreadSleepTime = 100;
  private int bufferSize = 10000;
  private String delimiter = "|";

  private PrintWriter out;

  private PARpcClientFacade client;

  public void init(DsConfiguration conf, DsMetaData metaData) {
    logger.info("Initializing handler: mode=" + getMode());
    super.init(conf, metaData); // always call 'super'

    // Initialize regular file
    try {
      out = new PrintWriter(new FileWriter(regularFileName));
    } catch (IOException ioe) {
      throw new ConfigException("Can't initialize file, " + regularFileName,
          ioe);
    }

    client = new PARpcClientFacade();
    // Initialize client with the remote Flume agent's host and port
    client.init(flumeHosts, clientType, hostSelector, backOff, maxBackOff,
        batchSize, connectTimeout, requestTimeout, sendThread,
        sendThreadSleepTime, bufferSize);

  }

  public Status transactionBegin(DsEvent e, DsTransaction tx) {
    logger
        .debug("-----------1----------------------------------transactionBegin");
    super.transactionBegin(e, tx);
    String eventFormat;
    eventFormat =
        String.format("Received begin tx event, numTx=%d : position=%s%n",
            numTxs, tx.getTranID());
    out.println(eventFormat);
    return Status.OK;
  }

  public Status operationAdded(DsEvent e, DsTransaction transaction,
      DsOperation operation) {
    super.operationAdded(e, transaction, operation);
    final Tx tx = new Tx(transaction, getMetaData(), getConfig());
    final TableMetaData tMeta =
        getMetaData().getTableMetaData(operation.getTableName());
    final Op op = new Op(operation, tMeta, getConfig());

    out.println("  Received operation: table='" + op.getTableName() + "'"
        + ", pos=" + op.getPosition() + " (total_ops= " + tx.getTotalOps()
        + ", buffered=" + tx.getSize() + ")" + ", ts=" + op.getTimestamp());

    if (isOperationMode()) {
      if (op.getOperationType().isDelete()) {
        processDel(tx, op);
      } else if (op.getOperationType().isUpdate()) {
        processUp(tx, op);
      } else {
        processOther(tx, op);
      }
    }
    return Status.OK;
  }

  @Override
  public Status transactionCommit(DsEvent e, DsTransaction transaction) {
    super.transactionCommit(e, transaction);

    numTxs++;

    Tx tx = new Tx(transaction, getMetaData(), getConfig());

    // in 'operation mode', all the operations would have been processed when
    // 'operationAdded' is called. In 'transaction mode', they are processed
    // when the commit event is received.
    if (!isOperationMode()) {
      for (Op op : tx) {
        if (op.getOperationType().isDelete()) {
          processDel(tx, op);
        } else if (op.getOperationType().isUpdate()) {
          processUp(tx, op);
        } else {
          processOther(tx, op); // process data...
        }
      }
    }

    out.println("  Received commit event, tx #" + numTxs + ": " + ", pos="
        + tx.getTranID() + " (total_ops= " + tx.getTotalOps() + ", buffered="
        + tx.getSize() + ")" + ", ts=" + tx.getTimestamp() + ")");
    out.flush();

    return Status.OK;
  }

  private void processDel(Tx currentTx, Op op) {
    logger.debug("processOp: tx=" + currentTx.getPosition() + ", op="
        + op.getPosition());
    numOps++;

    TableName tname = op.getTableName();
    TableMetaData tMeta = getMetaData().getTableMetaData(tname);

    String key = "";

    for (Iterator<ColumnMetaData> itr = tMeta.getKeyColumns().iterator(); itr
        .hasNext();) {
      ColumnMetaData cmd = itr.next();
      key += op.getColumn(cmd.getIndex()).getBeforeValue();
    }

    logger.debug("tname: =" + tname + ",keyval=" + key);

    StringBuffer sb = new StringBuffer();
    sb.append(tname);
    sb.append(delimiter);
    sb.append(op.getOperationType());
    sb.append(delimiter);
    sb.append(key);
    sb.append(delimiter);

    logger.debug("sb: =" + sb.toString());

    client.sendDataToFlume(sb.toString());
  }

  private void processOther(Tx currentTx, Op op) {
    // logger.debug("processOp: tx=" + currentTx.getPosition() + ", op=" +
    // op.getPosition() + ",OperationType=" + op.getOperationType());
    numOps++;

    TableName tname = op.getTableName();
    TableMetaData tMeta = getMetaData().getTableMetaData(tname);

    String key = "";

    for (Iterator<ColumnMetaData> itr = tMeta.getKeyColumns().iterator(); itr
        .hasNext();) {
      ColumnMetaData cmd = itr.next();
      key += op.getColumn(cmd.getIndex()).getAfterValue();
    }

    StringBuffer sb = new StringBuffer();
    sb.append(tname);
    sb.append(delimiter);
    sb.append(op.getOperationType());
    sb.append(delimiter);
    sb.append(key);
    sb.append(delimiter);

    int i = 0;
    for (Col c : op) {
      sb.append(tMeta.getColumnName(i) + "=" + c.getAfterValue());
      sb.append(delimiter);
      i++;
    }

    // logger.debug("sb: =" + sb.toString());
    client.sendDataToFlume(sb.toString());
  }

  private void processUp(Tx currentTx, Op op) {
    logger.debug("processOp: tx=" + currentTx.getPosition() + ", op="
        + op.getPosition() + ",OperationType=" + op.getOperationType());
    numOps++;

    TableName tname = op.getTableName();
    TableMetaData tMeta = getMetaData().getTableMetaData(tname);

    String key = "";

    for (Iterator<ColumnMetaData> itr = tMeta.getKeyColumns().iterator(); itr
        .hasNext();) {
      ColumnMetaData cmd = itr.next();
      key += op.getColumn(cmd.getIndex()).getAfterValue();
    }

    logger.debug("tname: =" + tname + ",keyval= " + key);

    StringBuffer sb = new StringBuffer();
    sb.append(tname);
    sb.append(delimiter);
    sb.append(op.getOperationType());
    sb.append(delimiter);
    sb.append(key);
    sb.append(delimiter);

    int i = 0;
    for (Col c : op) {
      if (c.isChanged()) {
        sb.append(tMeta.getColumnName(i) + "=" + c.getAfterValue());
        sb.append(delimiter);
      }
      i++;
    }

    logger.debug("sb: =" + sb.toString());
    client.sendDataToFlume(sb.toString());
  }

  public Status metaDataChanged(DsEvent e, DsMetaData meta) {
    logger.debug("metaDataChanged");
    super.metaDataChanged(e, meta);
    logger.debug("Received metadata event: " + e + "; current tables: "
        + meta.getTableNames().size());
    return Status.OK;
  }

  public void destroy() {
    logger.debug("destroy");
    out.println("Closing file... " + reportStatus());
    try {
      out.flush();
      out.close();
      client.cleanUp();
    } catch (Exception e) {

    }
    super.destroy();
  }

  public String reportStatus() {
    logger.debug("reportStatus");
    String s =
        "Status report: file=" + regularFileName + ", mode=" + getMode()
            + ", transactions=" + numTxs + ", operations=" + numOps;
    return s;
  }

  // Property setters methods.

  public void setRegularFileName(String filename) {
    logger.info("set regular filename: " + filename);
    this.regularFileName = filename;
  }

  public void setHosts(String hosts) {
    this.flumeHosts = hosts;
  }

  public void setClientType(String clientType) {
    this.clientType = clientType;
  }

  public void setHostSelector(String hostSelector) {
    this.hostSelector = hostSelector;
  }

  public void setBackOff(boolean backOff) {
    this.backOff = backOff;
  }

  public void setMaxBackOff(int maxBackOff) {
    this.maxBackOff = maxBackOff;
  }

  public void setDelimiter(String delimiter) {
    this.delimiter = delimiter;
  }

  public void setBatchSize(int batchSize) {
    this.batchSize = batchSize;
  }

  public void setConnectTimeout(int connectTimeout) {
    this.connectTimeout = connectTimeout;
  }

  public void setRequestTimeout(int requestTimeout) {
    this.requestTimeout = requestTimeout;
  }

  public void setSendThread(int sendThread) {
    this.sendThread = sendThread;
  }

  public void setSendThreadSleepTime(int sendThreadSleepTime) {
    this.sendThreadSleepTime = sendThreadSleepTime;
  }

  public void setBufferSize(int bufferSize) {
    this.bufferSize = bufferSize;
  }

}

class PARpcClientFacade {

  final private static Logger logger = LoggerFactory
      .getLogger(PARpcClientFacade.class);

  private RpcClient client;
  private String flumeHosts;
  private String clientType;
  private String hostSelector;
  private boolean backOff;
  private int maxBackOff;
  private int batchSize;
  private int connectTimeout;
  private int requestTimeout;

  public void init(String hosts, String clientType, String hostSelector,
      boolean backOff, int maxBackOff, int batchSize, int connectTimeout,
      int requestTimeout, int sendThread, int sendThreadSleepTime,
      int bufferSize) {
    this.flumeHosts = hosts;
    this.clientType = clientType;
    this.hostSelector = hostSelector;
    this.backOff = backOff;
    this.maxBackOff = maxBackOff;
    this.batchSize = batchSize;
    this.connectTimeout = connectTimeout;
    this.requestTimeout = requestTimeout;

    // Create the client with load balancing properties
    client = RpcClientFactory.getInstance(getProps());
  }

  public void sendDataToFlume(String data) {
    // Create a Flume Event object that encapsulates the data
    Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));

    try {
      client.append(event);
    } catch (EventDeliveryException e) {
      logger.error(e.getMessage());
      client = RpcClientFactory.getInstance(getProps());
    }
  }

  private Properties getProps() {
    Properties props = new Properties();
    props.put("client.type", clientType);
    props.put("host-selector", hostSelector);
    props.put("backoff", backOff);
    props.put("maxBackoff", maxBackOff);
    String[] singleHosts = flumeHosts.split(",");

    StringBuffer sb = new StringBuffer();

    for (int i = 0; i < singleHosts.length; i++) {
      sb.append("h" + (i + 1));
      sb.append(" ");
      props.put("hosts.h" + (i + 1), singleHosts[i]);
    }

    props.put("hosts", sb.toString().trim());
    props.put("batch-size", batchSize);
    props.put("connect-timeout", connectTimeout);
    props.put("request-timeout", requestTimeout);

    return props;
  }

  public void cleanUp() {
    // Close the RPC connection and stop send thead
    client.close();
    client = null;
  }

}